package com.hyt.it.ogt.ks.function;

import io.seata.common.util.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.util.Pair;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author 朱振宇
 * @since 2022-12-03
 */
@Slf4j
@Component
public class KsQueueHandleExecutor implements InitializingBean {

    private static final BlockingQueue<Pair<?, ProducerConsumeFunction>> taskQueue = new LinkedBlockingQueue<>(2048);

    private static final long WAIT_TIME = 50;

    private static final int BATCH_SIZE = 100;

    @Override
    public void afterPropertiesSet() {
        new KsExecutorThreadFactory("queueTask").newThread(new ExecuteTask()).start();
    }

    public static Queue<Pair<?, ProducerConsumeFunction>> getQueue() {
        return taskQueue;
    }

    final class ExecuteTask implements Runnable {
        @Override
        public void run() {
            List<Object> taskList;
            while (true) {
                try {
                    Object msg = taskQueue.take();
                    taskList = new ArrayList<>(BATCH_SIZE);
                    taskList.add(msg);
                    while (taskList.size() < BATCH_SIZE && Objects.nonNull((msg = taskQueue.poll(WAIT_TIME, TimeUnit.MILLISECONDS)))) {
                        taskList.add(msg);
                    }
                    if (CollectionUtils.isNotEmpty(taskList)) {
                        consume(taskList);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void consume(List<?> data) {
        try {
            data.forEach(vo -> ExecutorFactory.getThreadPoolExecutor().execute(() -> ((Pair<?, ProducerConsumeFunction>) vo).getSecond().apply(((Pair<?, ProducerConsumeFunction>) vo).getFirst())));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
